import com.alibaba.fastjson.JSON;
import com.yanhua.transaction.common.dto.TransactionGroupDto;
import com.yanhua.transaction.common.util.TransactionConst;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.*;

/**
 * 作为事务管理者，它需要：
 * 1. 创建并保存事务组
 * 2. 保存各个子事务在对应的事务组内
 * 3. 统计并判断事务组内的各个子事务状态，以算出当前事务组的状态（提交or回滚）
 * 4. 通知各个子事务提交或回滚
 */

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    // 事务组中的事务状态列表
    private static Map<String, List<TransactionConst.TransactionType>> transactionTypeMap = new HashMap<>();
    // 事务组是否已经接收到结束的标记
    private static Map<String, Boolean> isEndMap = new HashMap<>();
    // 事务组中应该有的事务个数
    private static Map<String, Integer> transactionCountMap = new HashMap<>();

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.add(ctx.channel());
    }

    /**
     * {groupId:List<子事务>}
     * 1. 接收创建事务组事件
     * 2. 接收子事务的注册事件
     * 3. 判断事务组的状态，如果该事务组中有一个事务需要回滚，那么整个事务组就需要回滚，反之，则整个事务组提交
     * 4. 通知所有客户端进行提交或回滚
     *
     * @param ctx
     * @param msg1
     * @throws Exception
     */

    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg1) throws Exception {
        System.out.println("接受数据:" + msg1.toString());
        TransactionGroupDto msg = JSON.parseObject(msg1.toString(), TransactionGroupDto.class);
        TransactionConst.TransactionCommand command = msg.getCommand();
        String groupId = msg.getGroupId();
        TransactionConst.TransactionType transactionType = msg.getTransactionType();
        Integer transactionCount = msg.getTransactionCount();
        TransactionConst.TransactionState state = msg.getState();
        if (Objects.equals(command, TransactionConst.TransactionCommand.create)) {
            // 创建事务组
            transactionTypeMap.put(groupId, new ArrayList<>());
        } else if (Objects.equals(command, TransactionConst.TransactionCommand.add)) {
            // 加入事务组
            transactionTypeMap.get(groupId).add(transactionType);
            if (Objects.equals(state, TransactionConst.TransactionState.end)) {
                //分布式事务结束标记
                isEndMap.put(groupId, true);
                transactionCountMap.put(groupId, transactionCount);
            }
            TransactionGroupDto.TransactionGroupDtoBuilder builder = TransactionGroupDto.builder();
            builder.groupId(groupId);
            Boolean aBoolean = isEndMap.get(groupId);
            // 如果已经接收到结束事务的标记，比较事务是否已经全部到达，如果已经全部到达则看是否需要回滚
            if (!Objects.isNull(aBoolean) && aBoolean
                    && Objects.equals(transactionCountMap.get(groupId), transactionTypeMap.get(groupId).size())) {
                //分布式事务结束
                if (transactionTypeMap.get(groupId).contains(TransactionConst.TransactionType.rollback)) {
                    //全部成功
                    builder.transactionType(TransactionConst.TransactionType.rollback);
                } else {
                    builder.transactionType(TransactionConst.TransactionType.commit);
                }
                sendResult(JSON.toJSONString(builder.build()));
            }

        }
    }

    private void sendResult(String result) {
        for (Channel channel : channelGroup) {
            System.out.println("发送数据111:" + result);
            channel.writeAndFlush(result);
        }
    }
}
